package org.apache.spark.doe

import org.apache.spark.{SecurityManager, SparkConf}
import org.apache.spark.rpc.{RpcEnv, ThreadSafeRpcEndpoint}

class Master(val rpcEnv: RpcEnv) extends ThreadSafeRpcEndpoint {

  //实现receive方法，接收异步消息
  override def receive: PartialFunction[Any, Unit] = {
    case "test" =>      println("接收到了测试消息")
    case RegisterWorker(workerId, host, port) => {
      println(s"接收到了注册消息$workerId, $host, $port")
    }

  }
}

object Master {

  def main(args: Array[String]): Unit = {
    //创建SparkConf对象
    val conf = new SparkConf
    //创建SecurityManager（安全管理器，对系统资源的访问进行检查和限制）
    val securityMgr = new SecurityManager(conf)
    //创建创建RpcEnv，并指定名称和id地址、端口等
    val rpcEnv = RpcEnv.create("SparkMaster", "localhost", 8888, conf, securityMgr)
    //创建Master RpcEndPoint
    val master = new Master(rpcEnv)
    //将Master RpcEndPoint的引用传入到setupEndpoint方法中，并指定名称
    val masterEndpoint = rpcEnv.setupEndpoint("Master", master)
    //自己想自己发送消息
    masterEndpoint.send("test")
    //将程序挂起，等待退出
    rpcEnv.awaitTermination()
  }
}
